package org.fing.edu.uy.esbadp.action.aggregator;

import static org.fing.edu.uy.esbadp.core.constants.EsbAdpPropertyConstants.AGGEGRATOR_ID_MESSAGE;
import static org.fing.edu.uy.esbadp.core.constants.EsbAdpPropertyConstants.AGGEGRATOR_TIME_STAMP;
import static org.fing.edu.uy.esbadp.core.constants.EsbAdpPropertyConstants.AGGEGRATOR_UUID_MESSAGE;
import static org.fing.edu.uy.esbadp.core.constants.EsbAdpPropertyConstants.TIME_OUT_MESSAGE;

import java.text.DateFormat;
import java.util.Date;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.logging.Level;

import org.fing.edu.uy.esbadp.assertion.AssertProperty;
import org.fing.edu.uy.esbadp.core.AdpFlowConstants;
import org.fing.edu.uy.esbadp.core.AdpFlowInterface;
import org.fing.edu.uy.esbadp.core.AdpFlowTree;
import org.fing.edu.uy.esbadp.core.constants.EsbAdpPropertyConstants;
import org.fing.edu.uy.esbadp.core.entity.GenericTreeNode;
import org.fing.edu.uy.esbadp.logger.LoggerADP;
import org.jboss.internal.soa.esb.assertion.AssertArgument;
import org.jboss.soa.esb.ConfigurationException;
import org.jboss.soa.esb.actions.AbstractActionPipelineProcessor;
import org.jboss.soa.esb.actions.ActionLifecycleException;
import org.jboss.soa.esb.actions.ActionProcessingException;
import org.jboss.soa.esb.helpers.ConfigTree;
import org.jboss.soa.esb.message.Message;

public class Aggregator extends AbstractActionPipelineProcessor {
	
	private static java.util.logging.Logger logger = LoggerADP.getLogger(Aggregator.class.getSimpleName());
	
    private Map<String, Long> aggregatedMessageMap = new ConcurrentHashMap<String, Long>();
    private TimeoutChecker _timeoutChecker = null;
    
    protected ConfigTree config;
    
    public Aggregator(){
    	
    }
    
    public Aggregator(ConfigTree config) throws ConfigurationException {        
        this.config = config;
	}
    
    /**
     * Initialise the action instance.
     * <p/>
     * This method is called after the action instance has been instantiated so that
     * configuration options can be validated.
     * 
     * @throws ActionLifecycleException for errors during initialisation.
     */
    public void initialise() throws ActionLifecycleException {
        _timeoutChecker = new TimeoutChecker();
        _timeoutChecker.start();
    }

    /**
     * Destroy the action instance.
     * <p/>
     * This method is called prior to the release of the action instance.  All
     * resources associated with this action instance should be released as the
     * instance will no longer be used.
     */
    public void destroy() throws ActionLifecycleException{
        _timeoutChecker.terminate();
        _timeoutChecker = null ;
    }
    
    /**
     * Processes an incoming message, aggregates messages in a set and returns a aggregated message
     * when isComplete() is satisfied. The aggregated messages are set as attachments to this message.
     * Next the message can be send to a transformer to do the second part of the aggregation which is 
     * to convert the attachement messages into to one message.
     * 
     * @param message
     * @return a aggregated message, or null if the aggregation has not been completed.
     * @throws ActionProcessingException
     */
    public Message process(Message message) throws ActionProcessingException {
    	
    	if(logger.isLoggable(LoggerADP.DEBUG_LEVEL)){
			logger.log(LoggerADP.DEBUG_LEVEL, "Process Action -> {0}", this.getClass().getName());
		}
    	
    	AssertArgument.isNotNull(message, "message");
		AssertProperty.isNotNull(message, AdpFlowConstants.PropertyFlowServices);
		
		//se obtiene el arbol con las acciones a realizar
		AdpFlowTree adpFlowTree = (AdpFlowTree) message.getProperties().getProperty(AdpFlowConstants.PropertyFlowServices);
		GenericTreeNode<AdpFlowInterface> treeNode = adpFlowTree.getCurrent();
		
		//como esta accion no tiene que decidir el proximo nodo es el primer hijo
		GenericTreeNode<AdpFlowInterface> nextCurrent = treeNode.getChildren().size() > 0 ? treeNode.getChildAt(0) : null;
		adpFlowTree.setCurrent(nextCurrent);
    	
    	try{
			String serialUuids = (String) message.getProperties().getProperty(AGGEGRATOR_UUID_MESSAGE);
			int index = serialUuids.indexOf(",");
			String nextSerialUuids =  serialUuids.substring(index+1);
			String serialUuid = serialUuids.substring(0, index);
			
			message.getProperties().setProperty(EsbAdpPropertyConstants.AGGEGRATOR_UUID_MESSAGE, nextSerialUuids);
			
			
			int idMessage = (Integer) message.getProperties().getProperty(AGGEGRATOR_ID_MESSAGE);
			//set timeStamp of response
			long timeStamp = new Date().getTime();
			message.getProperties().setProperty(AGGEGRATOR_TIME_STAMP, timeStamp);
			
			synchronized (aggregatedMessageMap) {
				//Get the messageMap with this uuId, or create it if not found
				if (aggregatedMessageMap.get(serialUuid) == null) {
					aggregatedMessageMap.put(serialUuid, timeStamp);
					if(logger.isLoggable(LoggerADP.DEBUG_LEVEL)){
						Object [] parameters = {serialUuid, idMessage};
						logger.log(LoggerADP.DEBUG_LEVEL, "Message Response : Id -> {0} and Relative Id -> {1}", parameters);
					}
				}else{
					message = null;
					if(logger.isLoggable(LoggerADP.DEBUG_LEVEL)){						
						logger.log(LoggerADP.DEBUG_LEVEL, "There is already a better response time and response that was returned for Message Id: ->" + serialUuid);
					}
				}
	        }
		}catch (Exception ex) {
			message = null;
			logger.log(Level.SEVERE, "Ignoring message, internal exception in Aggregator EsbAdp.", ex);
		}
		
        return message;
	}
    
    /**
     * If the aggregation process is complete then return true. This depends on the configuration. 
     * 
     * @param splitterTimeStamp The splitter timestamp.
     * @return True if the message is timed out, otherwise false.
     */
    private boolean isTimedOut(long splitterTimeStamp){
        long now = new Date().getTime();
        long expiration = splitterTimeStamp + TIME_OUT_MESSAGE;

        if ( now > expiration) {
        	if (logger.isLoggable(LoggerADP.DEBUG_LEVEL)) {
        		DateFormat dateFormat = DateFormat.getTimeInstance();
        		logger.log(LoggerADP.DEBUG_LEVEL, "message expired.");
        		logger.log(LoggerADP.DEBUG_LEVEL, "Current time=" + dateFormat.format(new Date(now)) + " message expiration=" + dateFormat.format(new Date(expiration)));
        	}
            return true;
        }
        
        return false;
    }
    
    /**
     * Check response timeout
     */
    class TimeoutChecker extends Thread {
        private final Lock terminateLock = new ReentrantLock() ;
        private final Condition terminateCondition = terminateLock.newCondition() ;
        private boolean terminated ;

        public void run() {

            boolean running = true;
            
            while(running) {
            	// is a snapshot so no lock needed
                for (Entry<String, Long> entry: aggregatedMessageMap.entrySet()) {
                	long timeStamp = entry.getValue();
                	String serialUuid = entry.getKey();
                    if (isTimedOut(timeStamp)) {
                        try {
                        	if (logger.isLoggable(LoggerADP.DEBUG_LEVEL)) {
                        		logger.log(LoggerADP.DEBUG_LEVEL, "Deleting message aggregation series: " + serialUuid);
                        	}
                        } catch(Throwable ex) {
                            logger.log(Level.SEVERE ,"Error delivering timed out aggregation message to Dead Letter Service.", ex);
                        } finally {
                        	synchronized (aggregatedMessageMap) {
                        		aggregatedMessageMap.remove(serialUuid);
                        	}
                        }
                    }
                }
                terminateLock.lock() ;
                try {
                    if (!terminated) {
                        terminateCondition.await(2500, TimeUnit.MILLISECONDS) ;
                    }
                    running = !terminated ;
                } catch (InterruptedException ie){
                } finally {
                    terminateLock.unlock() ;
                }
            }
        }

        public void terminate() {
            terminateLock.lock() ;
            try{
                terminated = true ;
                terminateCondition.signalAll() ;
            }
            finally{
                terminateLock.unlock() ;
            }
        }
    }
    
}
